package com.archive.core.executor;

import com.archive.core.config.ArchiveProperties;
import com.archive.core.processor.ArchiveProcessor;
import com.archive.core.statistics.ArchiveStatistics;
import com.archive.core.table.ArchiveColumn;
import com.archive.core.table.ArchiveTable;
import com.archive.core.util.ArchiveUtil;
import com.archive.core.util.SystemClock;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Data
@Slf4j
public abstract class AbstractArchiveExecutor implements ArchiveExecutor, OriginConnectionProvider {

    private final ArchiveProcessor archiveProcessor;
    private final ArchiveProperties archiveProperties;
    private OriginConnectionProvider originConnectionProvider;

    public AbstractArchiveExecutor(ArchiveProcessor archiveProcessor) {
        this.archiveProcessor = archiveProcessor;
        this.archiveProperties = archiveProcessor.getArchiveProperties();
    }

    @Override
    public void setOriginConnectionProvider(OriginConnectionProvider originConnectionProvider) {
        this.originConnectionProvider = originConnectionProvider;
    }

    @Override
    public Object queryData(ArchiveTable archiveTable, long startTime, Long archiveTimeStamp, Object maxPrimaryValue, Connection connection) throws SQLException {
        ResultSet resultSet = null;
        Statement statement = null;
        try {
            long start = SystemClock.now();
            // 主表判断超时
            if (ArchiveUtil.checkTimeout(archiveTable, archiveProcessor.getArchiveTables(), startTime, archiveProperties.getArchiveTimeout())) {
                log.info("table {} archive timeout, the archive task will be finished", archiveTable.getTableName());
                return null;
            }

            String sql = getQuerySql(archiveTable, archiveTimeStamp, maxPrimaryValue);

            Integer batchSize = archiveTable.getBatchSize();
            Map<String, List<Object>> extractDataMap = new HashMap<>();
            List<String> extractColumns = archiveTable.getExtractColumns();
            for (String extractColumn : extractColumns) {
                extractDataMap.put(extractColumn, new ArrayList<>(batchSize));
            }

            statement = connection.createStatement();
            statement.setFetchSize(archiveTable.getBatchSize());
            resultSet = statement.executeQuery(sql);
            List<Object[]> data = new ArrayList<>(batchSize);
            // 提取数据
            maxPrimaryValue = extractQueryData(archiveTable, maxPrimaryValue, resultSet, extractColumns, extractDataMap, data);

            archiveTable.setData(data);
            archiveTable.setExtractDataMap(extractDataMap);
            log.info("table {} queryData cost time：{}ms", archiveTable.getTableName(), SystemClock.now() - start);
            return maxPrimaryValue;
        } finally {
            ArchiveUtil.close(connection, statement, resultSet);
        }
    }

    /**
     * 提取数据
     */
    @Override
    public Object extractQueryData(ArchiveTable archiveTable, Object maxPrimaryValue, ResultSet resultSet, List<String> extractColumns,
                                   Map<String, List<Object>> extractDataMap, List<Object[]> data) throws SQLException {
        while (resultSet.next()) {
            String[] columnNames = archiveTable.getColumnNames();
            Object[] rowData = new Object[columnNames.length];
            for (int i = 0; i < columnNames.length; i++) {
                String columnName = columnNames[i];
                Object columnValue = resultSet.getObject(i + 1);
                rowData[i] = columnValue;

                if (extractColumns.contains(columnName)) {
                    extractDataMap.get(columnName).add(columnValue);
                }
            }
            data.add(rowData);
        }
        int size = data.size();
        if (size > 0) {
            maxPrimaryValue = data.get(size - 1)[archiveTable.getPrimaryKeyIndex()];
        }
        return maxPrimaryValue;
    }

    @Override
    public String getQuerySql(ArchiveTable archiveTable, Long archiveTimeStamp, Object maxPrimaryValue) {
        StringBuilder sql = new StringBuilder();
        String shortName = archiveTable.getShortName();
        boolean appendShortName = shortName != null && !shortName.trim().isEmpty();
        // 添加查询字段
        appendQueryColumns(archiveTable, sql, appendShortName, shortName);
        sql.append(" from ").append(archiveTable.getTableName());
        if (appendShortName) {
            sql.append(" ").append(shortName);
        }

        // 强制索引
        if (archiveTable.getIndexName() != null && !archiveTable.getIndexName().trim().isEmpty()) {
            sql.append(" force index(`").append(archiveTable.getIndexName()).append("`)");
        }

        // 解析archiveCondition
        String archiveCondition = archiveTable.getArchiveCondition();
        Properties properties = getProperties(archiveTable, archiveTimeStamp);
        archiveCondition = replacePlaceholders(archiveCondition, properties);
        sql.append(" where ").append(archiveCondition);
        if (maxPrimaryValue != null) {
            sql.append(" and ").append(archiveTable.getPrimaryKey()).append(" > ");
            appendColumnValue(sql, maxPrimaryValue);
            sql.setLength(sql.length() - 1);
        }
        sql.append(" order by ").append(archiveTable.getPrimaryKey());
        sql.append(" limit ").append(archiveTable.getBatchSize());

        return sql.toString();
    }

    /**
     * 添加查询字段，不同实现要查询的字段可能不同
     *
     * @param archiveTable    表信息
     * @param sql             sql语句
     * @param appendShortName 是否添加表别名
     * @param shortName       表别名
     */
    protected void appendQueryColumns(ArchiveTable archiveTable, StringBuilder sql, boolean appendShortName, String shortName) {
        List<ArchiveColumn> columns = archiveTable.getColumns();
        sql.append("select ");

        for (ArchiveColumn column : columns) {
            if (appendShortName) {
                sql.append(shortName).append(".");
            }
            sql.append(column.getColumnName()).append(",");
        }
        sql.deleteCharAt(sql.length() - 1);
    }

    private String replacePlaceholders(String archiveCondition, Properties properties) {
        Pattern pattern = Pattern.compile("#\\{\\s*(\\S+)\\s*}");
        Matcher matcher = pattern.matcher(archiveCondition);
        StringBuffer sb = new StringBuffer();
        while (matcher.find()) {
            String group = matcher.group();
            String value = properties.getProperty(group.substring(2, group.length() - 1).trim());
            matcher.appendReplacement(sb, value);
        }
        matcher.appendTail(sb);

        return sb.toString();
    }

    private Properties getProperties(ArchiveTable archiveTable, Long archiveTimeStamp) {
        Properties properties = new Properties();
        properties.put("archiveTimeStamp", archiveTimeStamp.toString());
        ArchiveTable parentTables = archiveTable.getParentTable();
        if (null != parentTables) {
            Map<String, List<Object>> parentExtractDataMap = parentTables.getExtractDataMap();
            if (parentExtractDataMap != null && !parentExtractDataMap.isEmpty()) {
                parentExtractDataMap.forEach((key, value) -> {
                    String extractKey = "extract." + key;
                    if (archiveTable.getExtractConditions().contains(extractKey) && !properties.contains(extractKey)) {
                        properties.put(extractKey, formatExtractData(value, false));
                    }
                    // 数字型id转字符串
                    String strKey = extractKey + ".string";
                    if (archiveTable.getExtractConditions().contains(strKey) && !properties.contains(strKey)) {
                        properties.put(extractKey + ".string", formatExtractData(value, true));
                    }
                });
            }
        }
        return properties;
    }

    private String formatExtractData(List<Object> value, boolean isStr) {
        StringBuilder sb = new StringBuilder();
        for (Object o : value) {
            if (isStr) {
                appendString(sb, o);
            } else {
                appendColumnValue(sb, o);
            }
        }
        return sb.substring(0, sb.length() - 1);
    }

    private void appendString(StringBuilder stringBuilder, Object value) {
        stringBuilder.append("'").append(value.toString().replace("\\", "\\\\")).append("'").append(",");
    }

    protected void appendColumnValue(StringBuilder stringBuilder, Object value) {
        if (value instanceof String) {
            stringBuilder.append("'").append(value.toString().replace("\\", "\\\\")).append("'").append(",");
        } else {
            stringBuilder.append(value).append(",");
        }
    }

    @Override
    public void doArchiveNormal(ArchiveTable archiveTable, String targetTableName, Connection connection,
                                Connection targetConnection, ArchiveStatistics archiveStatistics) throws SQLException {
        Statement insertStatement = null;
        Statement deleteStatement = null;
        try {
            if (targetConnection != null) {
                insertStatement = targetConnection.createStatement();
            } else {
                insertStatement = connection.createStatement();
            }
            deleteStatement = connection.createStatement();

            StringBuilder insertSql = new StringBuilder();
            int insertSqlClearPoint = getNormalInsertSql(archiveTable, targetTableName, insertSql);

            StringBuilder deleteSql = getDeleteSql(archiveTable);
            int deleteSqlClearPoint = deleteSql.length();

            createAndAddBatch(archiveTable, insertStatement, deleteStatement, insertSql, insertSqlClearPoint, deleteSql, deleteSqlClearPoint);

            // 执行批量插入和删除
            doExecuteBatch(archiveTable, archiveStatistics, insertStatement, deleteStatement);
        } finally {
            ArchiveUtil.closeStatement(deleteStatement);
            ArchiveUtil.closeStatement(insertStatement);
        }
    }

    protected int getNormalInsertSql(ArchiveTable archiveTable, String targetTableName, StringBuilder insertSql) {
        String[] columnNames = archiveTable.getColumnNames();
        insertSql.append("insert into ").append(targetTableName).append("(");
        for (String column : columnNames) {
            insertSql.append(column).append(",");
        }
        insertSql.setLength(insertSql.length() - 1);
        insertSql.append(") values ");
        return insertSql.length();
    }

    private void createAndAddBatch(ArchiveTable archiveTable, Statement insertStatement, Statement deleteStatement,
                                   StringBuilder insertSql, int insertSqlClearPoint, StringBuilder deleteSql, int deleteSqlClearPoint) throws SQLException {
        int batchSize = 0;
        Integer primaryKeyIndex = archiveTable.getPrimaryKeyIndex();
        for (Object[] datum : archiveTable.getData()) {
            // 添加字段值
            appendNormalInsertValue(insertSql, datum);

            appendColumnValue(deleteSql, datum[primaryKeyIndex]);

            batchSize++;
            if (batchSize % archiveProperties.getNormalInsertExecuteBatchSize() == 0) {
                addInsertBatch(insertStatement, insertSql, insertSqlClearPoint);
            }
            if (batchSize % archiveProperties.getDeleteExecuteBatchSize() == 0) {
                addDeleteBatch(deleteStatement, deleteSql, deleteSqlClearPoint);
            }
        }
        if (insertSql.length() > insertSqlClearPoint) {
            addInsertBatch(insertStatement, insertSql, insertSqlClearPoint);
        }
        if (deleteSql.length() > deleteSqlClearPoint) {
            addDeleteBatch(deleteStatement, deleteSql, deleteSqlClearPoint);
        }
    }

    protected void appendNormalInsertValue(StringBuilder insertSql, Object[] datum) {
        insertSql.append("(");
        for (Object value : datum) {
            appendColumnValue(insertSql, value);
        }
        insertSql.setLength(insertSql.length() - 1);
        insertSql.append("),");
    }

    protected void addDeleteBatch(Statement deleteStatement, StringBuilder deleteSql, int deleteSqlClearPoint) throws SQLException {
        deleteSql.setLength(deleteSql.length() - 1);
        deleteSql.append(")");
        deleteStatement.addBatch(deleteSql.toString());
        deleteSql.setLength(deleteSqlClearPoint);
    }

    protected void addInsertBatch(Statement insertStatement, StringBuilder insertSql, int insertSqlClearPoint) throws SQLException {
        insertSql.setLength(insertSql.length() - 1);
        insertStatement.addBatch(insertSql.toString());
        insertSql.setLength(insertSqlClearPoint);
    }

    private StringBuilder getDeleteSql(ArchiveTable archiveTable) {
        StringBuilder deleteSql = new StringBuilder("delete from ").append(archiveTable.getTableName()).append(" where ");
        deleteSql.append(archiveTable.getPrimaryKey()).append(" in (");
        return deleteSql;
    }

    protected StringBuilder getPreparedInsertSql(ArchiveTable archiveTable, String archiveTableName) {
        StringBuilder insertSql = new StringBuilder("insert into ").append(archiveTableName).append("(");
        StringBuilder valueSql = new StringBuilder(" values (");
        for (String columnName : archiveTable.getColumnNames()) {
            insertSql.append(columnName).append(",");
            valueSql.append("?,");
        }
        insertSql.deleteCharAt(insertSql.length() - 1);
        valueSql.deleteCharAt(valueSql.length() - 1);
        insertSql.append(")");
        valueSql.append(")");
        return insertSql.append(valueSql);
    }

    private void doExecuteBatch(ArchiveTable archiveTable, ArchiveStatistics archiveStatistics, Statement insertStatement,
                                Statement deleteStatement) throws SQLException {
        long currentTimeMillis = SystemClock.now();
        archiveStatistics.executeInsertStart(archiveTable);
        insertStatement.executeBatch();
        archiveStatistics.executeInsertEnd(archiveTable);
        log.info("table {} doArchive insertStatement execute cost time：{}ms", archiveTable.getTableName(), SystemClock.now() - currentTimeMillis);
        currentTimeMillis = SystemClock.now();
        archiveStatistics.executeDeleteStart(archiveTable);
        deleteStatement.executeBatch();
        archiveStatistics.executeDeleteEnd(archiveTable);
        log.info("table {} doArchive deleteStatement execute cost time：{}ms", archiveTable.getTableName(), SystemClock.now() - currentTimeMillis);
    }

    @Override
    public void doArchivePrepared(ArchiveTable archiveTable, String archiveTableName, Connection connection, Connection targetConnection,
                                  ArchiveStatistics archiveStatistics) throws SQLException {
        Statement insertStatement = null;
        Statement deleteStatement = null;
        try {
            StringBuilder insertSql = getPreparedInsertSql(archiveTable, archiveTableName);
            int insertSqlClearPoint = insertSql.length();
            StringBuilder deleteSql = getDeleteSql(archiveTable);
            int deleteSqlClearPoint = deleteSql.length();
            insertStatement = getPreparedInsertStatement(connection, targetConnection, insertSql);
            deleteStatement = connection.createStatement();
            int batchSize = 0;
            for (Object[] row : archiveTable.getData()) {
                batchSize++;
                // 添加字段值
                appendPreparedInsertValue(insertStatement, row, insertSql, batchSize, archiveProperties.getInsertIntoSelectBatchSize(), insertSqlClearPoint);

                appendColumnValue(deleteSql, row[archiveTable.getPrimaryKeyIndex()]);

                if (batchSize % archiveProperties.getDeleteExecuteBatchSize() == 0) {
                    addDeleteBatch(deleteStatement, deleteSql, deleteSqlClearPoint);
                }
            }
            if (insertSql.length() > insertSqlClearPoint) {
                appendPreparedRestInsertValue(insertStatement, insertSql, insertSqlClearPoint);
            }
            if (deleteSql.length() > deleteSqlClearPoint) {
                addDeleteBatch(deleteStatement, deleteSql, deleteSqlClearPoint);
            }
            // 执行批量插入和删除
            doExecuteBatch(archiveTable, archiveStatistics, insertStatement, deleteStatement);
        } finally {
            ArchiveUtil.closeStatement(insertStatement);
            ArchiveUtil.closeStatement(deleteStatement);
        }
    }

    protected abstract void appendPreparedRestInsertValue(Statement insertStatement, StringBuilder insertSql, int insertSqlClearPoint) throws SQLException;

    protected abstract void appendPreparedInsertValue(Statement insertStatement, Object[] row, StringBuilder insertSql, int batchSize, Integer executeBatchSize, int clearPoint) throws SQLException;

    protected abstract Statement getPreparedInsertStatement(Connection connection, Connection targetConnection, StringBuilder insertSql) throws SQLException;

    @Override
    public void handleDuplicateKeyException(ArchiveTable archiveTable, String targetTableName, Connection connection,
                                            Connection targetConnection, ArchiveStatistics archiveStatistics) throws SQLException {
        log.warn("table {} doArchive duplicate key, retry", archiveTable.getTableName());
        if (targetConnection == null) {
            targetConnection = connection;
        }
        String primaryKey = archiveTable.getPrimaryKey();
        Integer primaryKeyIndex = archiveTable.getPrimaryKeyIndex();
        StringBuilder selectSql = new StringBuilder("select ").append(primaryKey).append(" from ").append(targetTableName).append(" where ").append(primaryKey).append(" in (");
        List<Object[]> data = archiveTable.getData();
        data.forEach(row -> appendColumnValue(selectSql, row[primaryKeyIndex]));
        selectSql.setLength(selectSql.length() - 1);
        selectSql.append(")");

        StringBuilder deleteSql = getDeleteSql(archiveTable);
        int deleteSqlClearPoint = deleteSql.length();
        Map<Object, Object[]> dataMap = data.stream().collect(Collectors.toMap(row -> row[primaryKeyIndex], Function.identity(), (v1, v2) -> v1));

        try (Statement targetConnectionStatement = targetConnection.createStatement();
             ResultSet targetResultSet = targetConnectionStatement.executeQuery(selectSql.toString());
             Statement connectionStatement = connection.createStatement()) {
            int size = 0;
            while (targetResultSet.next()) {
                Object value = targetResultSet.getObject(1);
                if (dataMap.containsKey(value)) {
                    dataMap.remove(value);
                    appendColumnValue(deleteSql, value);
                    size++;
                    if (size % archiveProperties.getDeleteExecuteBatchSize() == 0) {
                        archiveStatistics.executeDeleteStart(archiveTable);
                        executeDeleteDuplicateKey(connectionStatement, deleteSql, deleteSqlClearPoint);
                        archiveStatistics.executeDeleteEnd(archiveTable);
                    }
                }
            }
            if (deleteSql.length() > deleteSqlClearPoint) {
                archiveStatistics.executeDeleteStart(archiveTable);
                executeDeleteDuplicateKey(connectionStatement, deleteSql, deleteSqlClearPoint);
                archiveStatistics.executeDeleteEnd(archiveTable);
            }
        }
        archiveTable.setData(new ArrayList<>(dataMap.values()));
        connection.commit();
        if (!archiveTable.getData().isEmpty()) {
            archiveProcessor.doArchive(archiveTable, targetTableName, archiveStatistics, this);
        }
    }

    private void executeDeleteDuplicateKey(Statement statement, StringBuilder deleteSql, int deleteSqlClearPoint) throws SQLException {
        deleteSql.setLength(deleteSql.length() - 1);
        deleteSql.append(")");
        statement.execute(deleteSql.toString());
        deleteSql.setLength(deleteSqlClearPoint);
    }

    /**
     * 检查同库归档表是否存在，不存在则创建
     *
     * @param tableName 表名
     */
    protected void checkArchiveTableSame(String tableName, String archiveTableName, Connection connection) throws SQLException {
        Statement statement = null;
        ResultSet resultSet = null;
        try {
            statement = connection.createStatement();

            StringBuilder sql = new StringBuilder("select TABLE_SCHEMA,TABLE_NAME from INFORMATION_SCHEMA.TABLES where TABLE_SCHEMA = '");
            sql.append(connection.getCatalog()).append("' and TABLE_NAME = '").append(archiveTableName).append("'");
            resultSet = statement.executeQuery(sql.toString());
            if (resultSet.next()) {
                // 检查表字段是否发生变化
                checkTableColumns(tableName, archiveTableName);
                return;
            }

            log.info("create archive table {}", archiveTableName);
            sql.setLength(0);
            sql.append("create table if not exists ").append(archiveTableName).append(" like ").append(tableName);
            statement.execute(sql.toString());
        } finally {
            ArchiveUtil.close(connection, statement, resultSet);
        }
    }

    /**
     * 检查不同库归档表是否存在，不存在则创建
     *
     * @param tableName 表名
     */
    protected void checkArchiveTableBackup(String tableName, String archiveTableName, Connection connection,
                                           Connection targetConnection) throws SQLException {
        Statement targetStatement = null;
        ResultSet targetResultSet = null;
        Statement statement = null;
        ResultSet resultSet = null;
        try {
            targetStatement = targetConnection.createStatement();
            String sql = "select TABLE_SCHEMA,TABLE_NAME from INFORMATION_SCHEMA.TABLES where TABLE_SCHEMA = '" +
                    targetConnection.getCatalog() + "' and TABLE_NAME = '" + archiveTableName + "'";
            targetResultSet = targetStatement.executeQuery(sql);
            if (targetResultSet.next()) {
                // 检查表字段是否发生变化
                checkTableColumns(tableName, archiveTableName);
                return;
            }

            // 原库表结构
            statement = connection.createStatement();
            resultSet = statement.executeQuery("show create table " + tableName);
            resultSet.next();
            String createTableSql = resultSet.getString("Create Table");
            if (!tableName.equals(archiveTableName)) {
                createTableSql = createTableSql.replace("CREATE TABLE `" + tableName + "`", "CREATE TABLE `" + archiveTableName + "`");
            }
            // 归档库创建归档表
            targetStatement.execute(createTableSql);
        } finally {
            ArchiveUtil.close(targetConnection, targetStatement, targetResultSet);
            ArchiveUtil.close(connection, statement, resultSet);
        }
    }

    private void checkTableColumns(String tableName, String archiveTableName) throws SQLException {
        if (archiveProperties.isCheckColumns()) {
            checkTableColumn(tableName, archiveTableName);
        }
    }

    /**
     * 检查更新原表与归档表字段
     *
     * @param tableName        原表名
     * @param archiveTableName 备份表名
     * @throws SQLException SQL异常
     */
    public abstract void checkTableColumn(String tableName, String archiveTableName) throws SQLException;

    /**
     * 表结构修改所需时长未知，使用原始数据库连接，数据源获取的连接可能导致超时
     *
     * @throws SQLException SQL异常
     */
    public Connection getConnection() throws SQLException {
        return originConnectionProvider.getConnection();
    }

    public Connection getTargetConnection() throws SQLException {
        return originConnectionProvider.getTargetConnection();
    }

    /**
     * 获取表字段
     *
     * @param catalog   数据库名称
     * @param tables    要获取字段的表，多个表使用","分隔： ‘table1’,‘table2’
     * @param statement 执行sql的statement
     * @return 表结构信息
     * @throws SQLException SQL异常
     */
    protected Map<String, Map<String, ArchiveColumn>> getColumns(String catalog, String tables, Statement statement) throws SQLException {
        ResultSet resultSet = null;
        Map<String, Map<String, ArchiveColumn>> tableColumns = new HashMap<>();
        try {
            StringBuilder sql = new StringBuilder("select TABLE_NAME,COLUMN_NAME,COLUMN_TYPE,CHARACTER_SET_NAME,COLLATION_NAME,IS_NULLABLE,COLUMN_COMMENT,COLUMN_DEFAULT " +
                    "from information_schema.COLUMNS where TABLE_SCHEMA = '")
                    .append(catalog).append("' and TABLE_NAME in (").append(tables).append(")");
            resultSet = statement.executeQuery(sql.toString());

            while (resultSet.next()) {
                String schemaTableName = resultSet.getString("TABLE_NAME");
                Map<String, ArchiveColumn> schemaColumnMap = tableColumns.computeIfAbsent(schemaTableName, k -> new HashMap<>());
                String columnName = resultSet.getString("COLUMN_NAME");
                String columnType = resultSet.getString("COLUMN_TYPE");
                String characterSetName = resultSet.getString("CHARACTER_SET_NAME");
                String collationName = resultSet.getString("COLLATION_NAME");
                String nullable = resultSet.getString("IS_NULLABLE");
                String columnComment = resultSet.getString("COLUMN_COMMENT");
                String columnDefault = resultSet.getString("COLUMN_DEFAULT");
                ArchiveColumn archiveColumn = new ArchiveColumn(columnName, columnType, nullable, characterSetName, collationName, columnDefault, columnComment);
                schemaColumnMap.put(columnName, archiveColumn);
            }
        } finally {
            ArchiveUtil.closeResultSet(resultSet);
        }

        return tableColumns;
    }

    /**
     * 检查并修改归档表字段
     *
     * @param tableName        原表名称
     * @param archiveTableName 备份表名称
     * @param tableColumns     原表字段
     * @param statement        执行sql的statement
     * @throws SQLException SQL异常
     */
    protected void checkAndModifyColumns(String tableName, String archiveTableName, Map<String, Map<String, ArchiveColumn>> tableColumns, Statement statement) throws SQLException {
        Map<String, ArchiveColumn> schemaColumnMap = tableColumns.get(tableName);
        StringBuilder changeSql = new StringBuilder("alter table `").append(archiveTableName).append("`");
        int length = changeSql.length();
        for (Map.Entry<String, ArchiveColumn> columnEntry : schemaColumnMap.entrySet()) {
            String column = columnEntry.getKey();
            ArchiveColumn originColumn = columnEntry.getValue();

            Map<String, ArchiveColumn> archiveColumnMap = tableColumns.get(archiveTableName);
            ArchiveColumn archiveColumn = archiveColumnMap.get(column);
            String columnType = originColumn.getColumnType();
            boolean changed = false;
            if (archiveColumn == null) {
                changeSql.append(" add column `");
                changed = true;
            } else if (!columnType.equals(archiveColumn.getColumnType())) {
                changeSql.append(" modify column `");
                changed = true;
            }
            if (changed) {
                changeSql.append(column).append("` ").append(columnType);
                if (columnType.contains("char") || columnType.contains("CHAR") || columnType.contains("text") || columnType.contains("TEXT")) {
                    String characterSetName = originColumn.getCharacterSetName();
                    String collationName = originColumn.getCollationName();
                    changeSql.append(" character set ").append(characterSetName).append(" collate ").append(collationName);
                }
                String columnDefault = originColumn.getColumnDefault();
                String columnComment = originColumn.getColumnComment();
                String nullable = originColumn.getNullable();
                if (columnDefault != null) {
                    changeSql.append(" default '").append(columnDefault).append("'");
                }
                if (nullable.equalsIgnoreCase("no")) {
                    changeSql.append(" not null");
                }
                if (columnComment != null) {
                    changeSql.append(" comment '").append(columnComment).append("'");
                }
                changeSql.append(",");
            }
        }
        if (changeSql.length() > length) {
            changeSql.setLength(changeSql.length() - 1);
            String sqls = changeSql.toString();
            long start = SystemClock.now();
            log.info("execute change sql: {}", sqls);
            statement.execute(sqls);
            log.info("execute change sql cost: {}", SystemClock.now() - start);
        }
    }

}
